package com.atguigu.chapter11;

import com.atguigu.chapter5.source.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Session;
import org.apache.flink.table.api.Slide;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.Tumble;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.junit.Test;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.lit;

/**
 * @ClassName: Flink11_Group_Window_Use
 * @Description:
 * @Author: kele
 * @Date: 2021/4/13 12:39
 *
 * 分组窗口的应用
 *
 *    滚动窗口
 *      .window( Tumble.over(lit(5).second()).on($("ts")).as("win") )
 *
 *    滑动窗口
 *      window( Slide.over( lit(5).second() ).every( lit(2).second() ).on($("ts")) .as("swin") )
 *
 *    会话窗口
 *      .window(Session.withGap(lit(2).second()).on($("ts")).as("sewin") )
 **/
public class Flink11_Group_Window_Table_Use {

    /**
     * 滚动窗口
     *  .window( Tumble.over(lit(5).second()).on($("ts")).as("win") )
     */
    @Test
    public void TumbleWindow() {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> waterSensorStream = env
                .fromElements(new WaterSensor("sensor_1", 1000L, 10),
                        new WaterSensor("sensor_1", 2000L, 20),
                        new WaterSensor("sensor_2", 3000L, 30),
                        new WaterSensor("sensor_1", 4000L, 40),
                        new WaterSensor("sensor_1", 5000L, 50),
                        new WaterSensor("sensor_2", 6000L, 60))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((element, recordTimestamp) -> element.getTs())
                );

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        Table table = tableEnv
                .fromDataStream(waterSensorStream,$("id"),$("ts").rowtime(),$("vc"));
/**
 * 使用table 窗口查询 数据
 */
        table
                .window( Tumble.over(lit(5).second()).on($("ts")).as("win") )
                .groupBy($("id"),$("win"))   //窗口分组
                .select($("id"),$("win").start().as("start"),$("vc").sum().as("vcsum") )
                .execute()
                .print();
    }


    /**
     *  滑动窗口
     *  window( Slide.over( lit(5).second() ).every( lit(2).second() ).on($("ts")) .as("swin") )
     */
    @Test
    public void SlidingWindow() {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> waterSensorStream = env
                .fromElements(new WaterSensor("sensor_1", 1000L, 10),
                        new WaterSensor("sensor_1", 2000L, 20),
                        new WaterSensor("sensor_2", 3000L, 30),
                        new WaterSensor("sensor_1", 4000L, 40),
                        new WaterSensor("sensor_1", 5000L, 50),
                        new WaterSensor("sensor_2", 6000L, 60))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((element, recordTimestamp) -> element.getTs())
                );

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        Table table = tableEnv
                .fromDataStream(waterSensorStream,$("id"),$("ts").rowtime(),$("vc"));

        table.
                window( Slide.over( lit(5).second() ).every( lit(2).second() ).on($("ts")) .as("swin") )
                .groupBy($("id"),$("swin"))
                .select( $("id"),$("swin").start().as("start"),$("vc").sum().as("vcsum") )
                .execute()
                .print();
    }

    /**
     * 会话窗口
     * .window(Session.withGap(lit(2).second()).on($("ts")).as("sewin") )
     */

    @Test
    public void SessionWindow() {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> waterSensorStream = env
                .fromElements(new WaterSensor("sensor_1", 1000L, 10),
                        new WaterSensor("sensor_1", 2000L, 20),
                        new WaterSensor("sensor_2", 3000L, 30),
                        new WaterSensor("sensor_1", 4000L, 40),
                        new WaterSensor("sensor_1", 5000L, 50),
                        new WaterSensor("sensor_2", 6000L, 60))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((element, recordTimestamp) -> element.getTs())
                );

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        Table table = tableEnv
                .fromDataStream(waterSensorStream,$("id"),$("ts").rowtime(),$("vc"));

        table
            .window(Session.withGap(lit(2).second()).on($("ts")).as("sewin") )
                .groupBy($("id"),$("sewin"))
                .select($("id"),$("sewin").start().as("start"),$("vc").sum().as("vcsum"))
                .execute()
                .print();
    }



}
